package channel

import (
	"channel/queue"
	"context"
	"sort"
	"sync"
	"time"
)

type DelayQueue[T Delayable] struct {
	comparer queue.Comparor[T]
	items    []T
	capacity int
	notFull  *sync.Cond
	notEmpty *sync.Cond
	mu       sync.Mutex
	pollCh   chan struct{}
	closed   bool
}

type Delayable interface {
	Delay() time.Duration // 更合理的定义
}

func NewDelayQueue[T Delayable](capacity int, comparer queue.Comparor[T]) *DelayQueue[T] {
	q := &DelayQueue[T]{
		comparer: comparer,
		capacity: capacity,
		items:    make([]T, 0, capacity),
		pollCh:   make(chan struct{}, 1),
	}
	q.notFull = sync.NewCond(&q.mu)
	q.notEmpty = sync.NewCond(&q.mu)
	return q
}

func (q *DelayQueue[T]) Enqueue(ctx context.Context, item T) error {
	q.mu.Lock()
	defer q.mu.Unlock()

	for q.isFull() && !q.closed {
		if err := q.waitWithContext(ctx, q.notFull); err != nil {
			return err
		}
	}

	if q.closed {
		return context.Canceled
	}

	q.items = append(q.items, item)
	sort.Slice(q.items, func(i, j int) bool {
		return q.comparer(q.items[i], q.items[j])
	})

	// 如果新添加的元素是最早到期的，通知等待的Dequeue
	if len(q.items) > 0 && q.comparer(item, q.items[0]) {
		select {
		case q.pollCh <- struct{}{}:
		default:
		}
	}

	q.notEmpty.Signal()
	return nil
}

func (q *DelayQueue[T]) Dequeue(ctx context.Context) (T, error) {
	q.mu.Lock()
	defer q.mu.Unlock()

	for q.isEmpty() && !q.closed {
		if err := q.waitWithContext(ctx, q.notEmpty); err != nil {
			var zero T
			return zero, err
		}
	}

	if q.closed {
		var zero T
		return zero, context.Canceled
	}

	now := time.Now()
	first := q.items[0]
	delay := first.Delay() - time.Since(now)

	if delay > 0 {
		// 需要等待
		q.mu.Unlock()

		timer := time.NewTimer(delay)
		defer timer.Stop()

		select {
		case <-ctx.Done():
			q.mu.Lock()
			return first, ctx.Err()
		case <-timer.C:
			q.mu.Lock()
		case <-q.pollCh:
			timer.Stop()
			q.mu.Lock()
			// 重新检查队列状态
			if q.isEmpty() || !q.comparer(first, q.items[0]) {
				continue
			}
		}
	}

	item := q.items[0]
	q.items = q.items[1:]
	q.notFull.Signal()
	return item, nil
}

func (q *DelayQueue[T]) waitWithContext(ctx context.Context, cond *sync.Cond) error {
	done := make(chan struct{})
	defer close(done)

	go func() {
		select {
		case <-ctx.Done():
			cond.Broadcast()
		case <-done:
		}
	}()

	cond.Wait()
	return ctx.Err()
}

func (q *DelayQueue[T]) isFull() bool {
	return len(q.items) == q.capacity
}

func (q *DelayQueue[T]) isEmpty() bool {
	return len(q.items) == 0
}

func (q *DelayQueue[T]) Close() {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.closed = true
	q.notFull.Broadcast()
	q.notEmpty.Broadcast()
}
